package each_webcode

import common.HbaseTools
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable
import scala.collection.mutable.HashMap
import scala.util.parsing.json.JSON;
object transferValueCode_to_Name_IBRD {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setAppName("IBRDdatachange").setMaster("local[*]") //集群运行去掉setMaster
    val sc = new SparkContext(sparkConf)
    sc.setLogLevel("Error")
    //val baseInfotable: String = new String("BaseMapInfos")
    //val detailTable: String = new String("DetailMapInfos")

    //val inpath = args(0)
    //val outpath = args(1)
    //val inpath = "/Users/kelisiduofu/IdeaProjects/zhiwang/InternationalData/makeChange/src/main/resources/sub_IBRD_Value.json"
    //val outpath = "/Users/kelisiduofu/IdeaProjects/zhiwang/InternationalData/makeChange/src/main/resources/result"

    //    val inpath="hdfs://node01:8020/tmp/InternationalData_source/InternationalData.IBRD_Value.json"
    //临时跑的教育数据
    val inpath = "file:///Users/kelisiduofu/IdeaProjects/zhiwang_file/jiaoyu/EducationStatistics/part-00000"
    val outpath = "file:///Users/kelisiduofu/IdeaProjects/zhiwang_file/jiaoyu/EducationStatistics_1"
//    val inpath = "file:///Users/kelisiduofu/IdeaProjects/zhiwang_file/jiaoyu/WorldDevelopmentIndicators/part-00000"
//    val outpath = "file:///Users/kelisiduofu/IdeaProjects/zhiwang_file/jiaoyu/WorldDevelopmentIndicators_1"
    val jsonStr = sc.textFile(inpath)

    val lines: RDD[Option[Any]] = jsonStr.map(one_line => JSON.parseFull(one_line))

    val pre_rdd = lines.map {
            //      r =>
            //        r match {
        case Some(m: Map[String, Any]) =>
                //{
            val my_map: HashMap[String, String] = new HashMap[String, String]()
            m.foreach(t => {
              //my_map += (t._1 -> t._2.toString.replace("\n", ""))
              my_map += (t._1 -> t._2.toString)
              })
            my_map("_id") = my_map("_id").substring(12, 36)
            (scala.util.Random.nextInt(200).toString + "_" + "IBRD_" + my_map("SourceCode"), my_map)
        //("IBRD_" + my_map("SourceCode"), my_map)
        //          }
        //        }
    }

    //val result_rdd: RDD[String] = pre_rdd.groupByKey(10).map {
            val result_rdd: RDD[String] = pre_rdd.groupByKey(400).map {
            case (k: String, v) => {
                val arr = k.split("_")
                //val k1=arr(0)+"_"+arr(1)
                 val k1 = arr(1) + "_" + arr(2)
                val tools = new HbaseTools()
                val connection = tools.getConnection()
                //dealTypes:map(kbase表的列名,列对应type类型)
                val dealTypes: Map[String, String] = tools.getFamily_column_value(connection, "BaseMapInfos", k1, "dealTypes").toMap
                //map(kbase列名,value表列名)
                val map01: Map[String, String] = tools.getFamily_column_value(connection, "BaseMapInfos", k1, "map01").toMap
                //map(conceptId,map(detailId,DetailName))
                //map("WebInfo",map(SourceCode,值))
                //map("WebInfo",map(SourceName,值))
                //map("WebInfo",map(WebCode,值))
                //map("WebInfo",map(WebName,值))
                val map04 = new mutable.HashMap[String, mutable.Map[String, String]]()
                val familys = tools.getFamilys_list(connection, "DetailMapInfos", k1)
                for (family: String <- familys) {
                        //列名,列值
                val tmp: mutable.HashMap[String, String] = tools.getFamily_column_value(connection, "DetailMapInfos", k1, family)
                    map04.put(family, tmp)
                }

            //map(detailname,detailId)
                val PU_timeMap: Map[String, String] = tools.getFamily_column_value(connection, "BaseMapInfos", "public", "public_Time").toMap
        //make_Dic,map(detailname,detailId)
                val PU_country: Map[String, String] = tools.getFamily_column_value(connection, "BaseMapInfos", "public", "public_country").toMap

                connection.close()
                val result: Iterable[String] = v.map {
                keys_values => {
                val result_map = new mutable.HashMap[String, String]()
                val map03 = keys_values
                dealTypes.keys.foreach {
                        key =>
                dealTypes(key) match {
                        case "type01" => { //map01+map03
                if (map01.contains(key) && map03.contains(map01(key))) {
                        //SourceCode=29,43,2   Year Time
                        //29:Time格式为YR2015
                        if (map03("SourceCode") == "29" && key == "Year" && map03(map01(key)).length == 6)
                                result_map.put(key, map03(map01(key)).split("R")(1))
                                //29:Time格式为YR2015-28
                                else if (map03("SourceCode") == "29" && key == "Year" && map03(map01(key)).length == 9)
                                result_map.put(key, map03(map01(key)).split("-")(0).split("R")(1))
                                //54:Time格式为YR1990Q1  Year取1008
                                else if (map03("SourceCode") == "43" && key == "Year" && map03(map01(key)).length == 8) {
                                result_map.put(key, map03(map01(key)).split("R")(1).split("Q")(0))
                                //此时加入Period字段,取Q1
                                result_map.put("Period", "Q" + map03(map01(key)).split("R")(1).split("Q")(1))
                                }
                                //或许有YR2015这样的格式
                                else if (map03("SourceCode") == "43" && key == "Year" && map03(map01(key)).length == 6) {
                                result_map.put(key, map03(map01(key)).split("R")(1))
                                }
                                //2:Time格式为YR1990
                                else if (map03("SourceCode") == "2" && key == "Year" && map03(map01(key)).length == 6) {
                                result_map.put(key, map03(map01(key)).split("R")(1))
                                }
                                //2:Time格式为YR1990Q1 Year取1008
                                else if (map03("SourceCode") == "2" && key == "Year" && map03(map01(key)).length == 8) {
                                result_map.put(key, map03(map01(key)).split("R")(1).split("Q")(0))
                                //此时加入Period字段,取Q1
                                result_map.put("Period", "Q" + map03(map01(key)).split("R")(1).split("Q")(1))
                                }
                                else if (map03("SourceCode") == "16" && key == "Year" && map03(map01(key)).length == 6) {
                                result_map.put(key, map03(map01(key)).split("R")(1))
                                }
                                else
                                result_map.put(key, map03(map01(key)))
                        }
                }
      case "type02" => {
              result_map.put(key, map04("WebInfo")(map01(key)))
              }
      case "type05" => {
              if (map01.contains(key) && map03.contains(map01(key)) && map04.contains(map01(key)) && map04(map01(key)).contains(map03(map01(key)))) {
                      //SourceCode=54,中的Year含有Q季度,切掉后面的Q季度,只保留年
                      if (map03("SourceCode") == "54" && key == "Year") {
                              result_map.put(key, map04(map01(key))(map03(map01(key))).split("Q")(0))
                              result_map.put("Period", "Q" + map04(map01(key))(map03(map01(key))).split("Q")(1))
                              }
                              //SourceCode=54,格式为2013 Targe或2013,切掉后面的Q季度,只保留年
                              else if (map03("SourceCode") == "34" && key == "Year")
                              result_map.put(key, map04(map01(key))(map03(map01(key))).split(" ")(0))
                              else
                              result_map.put(key, map04(map01(key))(map03(map01(key))))
                      }
              }
      //SourceCode=15 20 22 23 67 68 76 77
      //Year Time 中的年度
      //Priod Time 中的季度
      case "type01_bird" => {
              if (map03(map01("Period")).length == 8) {
                      val time = map03(map01("Period"))
                      result_map.put("Year", time.substring(2, 6))
                      result_map.put("Period", time.substring(6, 8))
                      }

              if (map03(map01("Period")).length == 6) {
                      val time = map03(map01("Period"))
                      result_map.put("Year", time.substring(0, 4))
                      result_map.put("Period", time.substring(4, 6))
                      }

              if (map03(map01("Period")).length == 7) {
                      val time = map03(map01("Period"))
                      result_map.put("Year", time.substring(0, 4))
                      result_map.put("Period", time.substring(4, 7))
                      }
              if (map03(map01("Period")).length == 4) {
                      val time = map03(map01("Period"))
                      result_map.put("Year", time)
                      result_map.put("Period", "")
                      }
              }
      case "type02_bird" => {
              if (map01.contains(key) && map03.contains(map01(key)) && map04.contains(map01(key)) && map04(map01(key)).contains(map03(map01(key)))) {
                      result_map.put(key, "Version:" + map04(map01(key))(map03(map01(key))))
                      }
              }
      //处理map03 key:'Receiving Countries ',多了一个空格,ReporterCode
      case "type03_bird" => { //map01+map03
              if (map01.contains(key) && map03.contains(map01(key) + " ")) {
                      result_map.put(key, map03(map01(key) + " "))
                      }
              }
      //处理map03 key:'Receiving Countries ',Reporter
      case "type04_bird" => { //map01+map03
              if (map01.contains(key) && map03.contains(map01(key) + " ") && map04.contains(map01(key)) && map04(map01(key)).contains(map03(map01(key) + " "))) {
                      result_map.put(key, map04(map01(key))(map03(map01(key) + " ")))
                      }
              }
//处理SourceCode=54,原为Year Time Time,改为Year Time Time中Q前面的字段
}
              }

      //生成TimeSort字段，
      if (result_map.contains("Year") && result_map.contains("Period") && result_map("Period").trim != "") {
              if (PU_timeMap.contains(result_map("Period").trim)) {
                      result_map.put("TimeSort", result_map("Year") + PU_timeMap(result_map("Period").trim))
                      } else {
                      result_map.put("TimeSort", result_map.getOrElse("Year", ""))
                      }
              } else {
              result_map.put("TimeSort", result_map.getOrElse("Year", ""))
              }

      //生成FuzzyIndicator字段
      if (result_map.contains("Indicator") && result_map("Indicator") != "") {
              result_map.put("FuzzyIndicator", result_map("Indicator"))
              //生成IndicatorLength字段
              result_map.put("IndicatorLength", result_map("Indicator").length.toString)
              }
      //生成FuzzyReporter字段
      if (result_map.contains("Reporter") && result_map("Reporter") != "") {
              if (PU_country.contains(result_map("Reporter"))) {
                      result_map.put("FuzzyReporter", result_map("Reporter") + "，" + PU_country(result_map("Reporter")))
                      } else {
                      result_map.put("FuzzyReporter", result_map("Reporter"))
                      }
              }

      val MetaData: Array[String] = Array("WebCode", "SourceCode", "ReporterCode", "Reporter", "ParternerCode", "Parterner", "Year", "Period",
              "IndicatorCategory", "Indicator", "Dim1Type", "Dim1", "Dim2Type", "Dim2", "Dim3Type", "Dim3",
              "Dim4Type", "Dim4", "Dim5Type", "Dim5", "Dim6Type", "Dim6", "Dim7Type", "Dim7", "Dim8Type", "Dim8")
      val DIMTYPEMETADATA: Array[String] = Array("Dim1Type", "Dim2Type", "Dim3Type", "Dim4Type", "Dim5Type", "Dim6Type", "Dim7Type", "Dim8Type")
      val DIMMETADATA: Array[String] = Array("Dim1", "Dim2", "Dim3", "Dim4", "Dim5", "Dim6", "Dim7", "Dim8")

      //生成DIMTYPEMETADATA:各Dim1Type值的组合,存在的为"值，"不存在的去掉
              val dimtypemetadata = DIMTYPEMETADATA.flatMap(result_map.getOrElse(_, "<->") + "，").mkString("").replace("<->，", "")

      //生成DIMMETADATA：各Dim组合，存在的为"值，"不存在的去掉
              val dimemetadata = DIMMETADATA.flatMap(onecel => {
              result_map.getOrElse(onecel, "<->") + "，"
              }).mkString("").replace("<->，", "")
      //生成MetaData：部分字段值的组合,存在的为"值，"不存在的去掉
              val metadata = MetaData.flatMap(result_map.getOrElse(_, "<->") + "，").mkString("").replace("<->，", "")
      result_map.put("DIMTYPEMETADATA", dimtypemetadata)
      result_map.put("DIMMETADATA", dimemetadata)
      result_map.put("MetaData", metadata)
      //val kkl:RDD[String] = sc.parallelize(().toString)
      //将map(key -> value)间的分隔符改成\t
      result_map.mkString("(<>)").replace(" -> ", "\t")
      }
}

                val res = result.mkString("\n")
                res
                            }
            }

    //val result_rdd: RDD[String]

    // 按照换行符切割
            val lines2: RDD[String] = result_rdd.flatMap(lines => {
            lines.split("\n")
            })

    val pangpang: RDD[String] = lines2.map(line => {
            // 按照\(<>\)切割
                    val strs: Array[String] = line.split("\\(<>\\)")

            val m = new mutable.HashMap[String, String]()

            strs.map(x => {
                    // 按照\t切割
                            val s = x.split("\t")
                    // 长度不为2,放入空串
                    if (s.length < 2) {
                            m.put(s(0), "")
                            }
                            else
                            // 长度为2,正常放入集合中
                            m.put(s(0), s(1))
                    })
            val res1: Map[String, String] = m.toMap
            // 过滤掉为空串的
                    val res2: Map[String, String]= res1.filter(x => {
                    var flag: Boolean = true
                    // 如果集合中不存在ObservationValue所对应的value值,或者value值为"",则返回false,过滤掉
                    if (m.getOrElse("ObservationValue", "").equals("") || m.getOrElse("Indicator", "").equals("") || m.getOrElse("Reporter", "").equals(""))
                            flag = false
                    flag
                    })
            // 将集合进行拼接
            res2.mkString("(<>)").replace(" -> ", "\t")
            })


//    pangpang.filter(!_.equals("")).saveAsTextFile(outpath)
    pangpang.filter(!_.equals("")).repartition(1).saveAsTextFile(outpath)
    sc.stop()
  }
}
